package com.lvmama.monkey.common.utils.JDBCUtils

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.DataFrame

import scala.collection.mutable.ArrayBuffer



/**
  * Created by hejing on 2017/10/11.
  */
object ConnetionPool {

  def getProperties(): Properties = {
    val properties = new Properties()
    properties.load(ConnetionPool.getClass.getResourceAsStream("/db-config.properties"))
    properties
  }

  def  getConnection(): Connection = {
    val properties = getProperties()
    val user = properties.getProperty("jdbc.username")
    val password = properties.getProperty("jdbc.password")
    val url = properties.getProperty("jdbc.url")
    Class.forName(properties.getProperty("jdbc.driverClassName")).newInstance()
    val connection: Connection = DriverManager.getConnection(url, user, password)
    connection
  }

  def upsert(iterator: Iterator[Array[String]],sql:String) = {
    val count = sql.length - sql.replace("?","").length
    var ps: PreparedStatement = null
    val conn: Connection = getConnection()
    try{
      iterator.foreach(data => {
        ps = conn.prepareStatement(sql)
        for(i <- 1 to count ) ps.setString(i,if(i > count /2 )data(i - count/2 -1) else data(i-1))
        ps.executeUpdate()
      })
    }catch{
      case e:Exception => e.printStackTrace()
    }finally {
      if(null != ps) ps.close()
      if(null != conn) conn.close()
    }
  }

  def upsertDF2Mysql(df: DataFrame,sql: String) = {
    df.show(false)
    val result: RDD[Array[String]] = df.rdd.map{
      r => {
        val arr = ArrayBuffer[String]()
        for (i <- 0 to r.length - 1){
          arr += r.get(i).toString
        }
        arr.toArray
      }
    }
    result.foreachPartition(x => upsert(x,sql))
  }

  def delFromMysql(sql:String): Unit ={
    var ps: PreparedStatement = null
    val conn: Connection = getConnection()
    try{
      ps = conn.prepareStatement(sql)
      ps.executeUpdate()
    }catch{
      case e:Exception => e.printStackTrace()
    }finally {
      if(null != ps) ps.close()
      if(null != conn) conn.close()
    }
  }

}
